package streamingStudy.testStreaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object DStreamFromQueseStream {

  def main(args: Array[String]): Unit = {
    //1.创建Spark配置信息
    val sparkConf = new SparkConf().setAppName("DStreamFromQueseStream").setMaster("local[*]")
    //2.初始化SparkStreamingContext
    val streamingContext = new StreamingContext(sparkConf, Seconds(4))
    //3.创建RDD队列
    val rddQueue = new mutable.Queue[RDD[Int]]()
    //4.创建QueueInputDStream
    val inputStream = streamingContext.queueStream(rddQueue, oneAtATime = false)
    //5.处理队列中的RDD数据
    val mappedStream = inputStream.map((_, 1))
    val reduceStream = mappedStream.reduceByKey(_ + _)
    //6.打印结果
    reduceStream.print()
    //7.启动任务
    streamingContext.start()
    //8.循环创建并向RDD队列中放入RDD
    for (i <- 1 to 5){
      rddQueue += streamingContext.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }
    streamingContext.awaitTermination()

  }

}
